package cn.xqapp.GatewayServer;

import cn.xqapp.core.BaseServer;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import org.apache.commons.lang3.time.StopWatch;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class GatewayServer extends BaseServer {

    protected int healthCheckIntervalSecs = 10;
    protected ScheduledExecutorService healthCheckerScheduler;
    protected List<GatewayServerHandler> serverHandlerList;

    protected StopWatch stopWatch;

    public GatewayServer(int port){
        super(port);

        serverHandlerList = new ArrayList<>();
        super.setChannelHandlerSupplier(this::gatewayServerHandlerSupplier);

        this.onBeforeRun(this::startHealthChecker);
        this.onBeforeShutdown(this::stopHealthChecker);

        stopWatch = new StopWatch();
        stopWatch.start();
    }

    private ChannelHandler gatewayServerHandlerSupplier(){
        return new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) {
                GatewayServerHandler serverHandler = new GatewayServerHandler();
                ch.pipeline().addLast(
                        new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4),
                        new LengthFieldPrepender(4),
                        serverHandler);

                synchronized (serverHandlerList){
                    serverHandlerList.add(serverHandler);
                }
            }
        };
    }

    public void startHealthChecker(){
        // try stop first
        stopHealthChecker();

        stopWatch.reset();
        stopWatch.start();

        healthCheckerScheduler = Executors.newScheduledThreadPool(1);
        healthCheckerScheduler.scheduleAtFixedRate(this::checkHealth, 0, this.healthCheckIntervalSecs, TimeUnit.SECONDS);
    }

    public void stopHealthChecker(){
        stopWatch.stop();
        if(healthCheckerScheduler != null){
            healthCheckerScheduler.shutdown();
        }
    }

    public void checkHealth(){
        stopWatch.stop();
        float totalSpanSecs = stopWatch.getNanoTime() / 1000000000f;

        // collect statics
        long totalConn = 0;
        long totalMsg = 0;
        long totalErr = 0;

        synchronized (serverHandlerList){
            for(GatewayServerHandler serverHandler:serverHandlerList){
                totalConn += serverHandler.getClientConnectionNum();
                totalMsg += serverHandler.getClientMessageNum();
                totalErr += serverHandler.getClientErrorNum();

                serverHandler.resetCounter();
            }
        }

        float ops = 0.0f;
        if(totalSpanSecs > 0){
            ops = totalMsg / totalSpanSecs;
        }

        System.out.println(String.format("[%s] CON %d MSG %d TIME %.2f OPS %.2f LAG N/A ERR %d",
                "SERVER", totalConn, totalMsg, totalSpanSecs, ops, totalErr));

        stopWatch.reset();
        stopWatch.start();
    }
}